package com.atguigu.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/10/27 13:57
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private ValueState<Boolean> isFirst;
    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 建立到Phoenix的连接
        String url = Constant.PHOENIX_URL;
        conn = JdbcUtil.getPhoenixConnection(url);
        
        isFirst = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirst", Boolean.class));
        
        redisClient = RedisUtil.getRedisClient();
        
    }
    
    @Override
    public void close() throws Exception {
        // 关闭连接
        if (conn != null) {
            conn.close();
        }
        
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context ctx) throws Exception {
        // 1. 建表
        checkTable(value);
        // 2. 写数据
        writeData(value);
        
        // 3. 更新缓存
        // 优雅: 更新缓存
        updateCache(value);
        
        // 粗暴: 直接删除对应的缓存
        //        delCache(value);
        
    }
    
    private void delCache(Tuple2<JSONObject, TableProcess> value) {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        String key = tp.getSink_table() + ":" + data.getLong("id");
        redisClient.del(key);
    }
    
    private void updateCache(Tuple2<JSONObject, TableProcess> value) {
        // key
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        String key = tp.getSink_table() + ":" + data.getLong("id");
        
        // 这次操作是更新维度, 还要去redis中存在这个缓存
        if ("update".equals(tp.getOperate_type()) && redisClient.exists(key)) {
            // value
            // 应该把字段名, 全部大写
            JSONObject upperDim = new JSONObject();
            for (Map.Entry<String, Object> entry : data.entrySet()) {
                upperDim.put(entry.getKey().toUpperCase(), entry.getValue());
            }
            
            redisClient.set(key, upperDim.toString());
        }
        
    }
    
    private void writeData(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        // 拼接插入语句 todo
        // upsert into t(a, b, c) values(?, ?, ?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(tp.getSink_table())
            .append("(")
            .append(tp.getSink_columns())
            .append(")values(")
            .append(tp.getSink_columns().replaceAll("[^,]+", "?"))
            .append(")");
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 给占位符赋值 todo  根据列名去data中取值
        String[] cs = tp.getSink_columns().split(",");
        for (int i = 0; i < cs.length; i++) {
            String columnName = cs[i];
            Object v = data.get(columnName);
            ps.setString(i + 1, v == null ? null : v.toString());
        }
        // 执行sql
        ps.execute();
        conn.commit();
        ps.close();
    }
    
    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        /*
        在Phoenix中建表
        执行建表语句
        
        create table if not exists t(id varchar, name varchar, age varchar, constraint pk primary key(id, name))
         */
        if (isFirst.value() == null) {
            TableProcess tp = value.f1;
            
            StringBuilder sql = new StringBuilder();
            sql
                .append("create table if not exists ")
                .append(tp.getSink_table())
                .append("(")
                .append(tp.getSink_columns().replaceAll(",", " varchar, "))
                .append(" varchar, constraint pk primary key(")
                .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                .append("))")
                .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());
            System.out.println(sql.toString());
            
            PreparedStatement ps = conn.prepareStatement(sql.toString());
            ps.execute();
            
            conn.commit();
            ps.close();
            
            isFirst.update(true);
            
        }
    }
}
